package net.bw.realtime.tmall.dws.job;
import net.bw.realtime.tmall.dws.funcion1.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import net.bw.realtime.tmall.common.utils.KafkaUtil;
import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import net.bw.realtime.tmall.dwd.function.TransactionEventBeanMapFunction;
import java.time.Duration;

/**
 * @Author: liuyawei
 * @Date: 2025/5/27
 */

public class TmallLogAPIWindow {

	public static void main(String[] args) throws Exception {

		// 1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2.数据源-source
		DataStream<String> logstream = KafkaUtil.consumerKafka(env, "tmall_log_test_abnormal");
//		pagestream.print();

		// 3.数据转换-transformation
		handle(logstream);
//		resultStream.print();

		// 5.触发执行-execute
		env.execute("JtpTrafficPageViewMinuteWindowDwsJob");
	}


	private static void handle(DataStream<String> logstream) {

		//s1.将流中每条日志数据封装成实体类Bean对象
		DataStream<TransactionEventBean> beanStream = logstream.map(
				new TransactionEventBeanMapFunction()
		);


		//s2-事件事件字段和水位线
		DataStream<TransactionEventBean> timeStream = beanStream.assignTimestampsAndWatermarks(
				WatermarkStrategy
						//允许乱序数据最大时间，如果是0，不考虑乱序迟到数据
						.<TransactionEventBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
						.withTimestampAssigner(
								new SerializableTimestampAssigner<TransactionEventBean>() {
									@Override
									public long extractTimestamp(TransactionEventBean TransactionEventBean, long l) {
										return TransactionEventBean.getTs();
									}
								}
						)
		);

		//s3-实时指标计算
		//todo a 实时交易额统计
		ToTradeGMVWindowFunction.toTradeGMV(timeStream);

		//todo b 实时订单量统计
		ToOrderCntWindowFunction.toOrderCnt(timeStream);

		//todo c 热门商品排行
		ToHotSkuWindowFunction.toHotSku(timeStream);

		//todo d 地区分布统计
		ToAreaDisWindowFunction.toAreaDistribution(timeStream);

		//todo e 异常交易监控 -- 保存至clickhouse
		CEPWindowFunction.toCepTrade(timeStream);

		//todo f 设备分布统计
		ToDeviceDisWindowFunction.toDeviceDistribution(timeStream);

	}

}


